package io.openkunlun.javadsl.client;

import com.google.protobuf.Empty;
import io.grpc.ConnectivityState;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import io.openkunlun.javadsl.DaprException;
import io.openkunlun.javadsl.v1.DaprCommonProtos;
import io.openkunlun.javadsl.v1.DaprGrpc;
import io.openkunlun.javadsl.v1.DaprProtos;

import java.io.Closeable;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

public final class DaprClient implements Closeable {

    public static final String DEFAULT_HOST = "127.0.0.1";
    public static final int DEFAULT_PORT = 12345;
    public static final int DEFAULT_TIMEOUT = 30 * 1000;
    public static final int DEFAULT_MAX_MESSAGE_BYTES = 16 * 1024 * 1024; // 16M
    public static final int DEFAULT_MAX_METADATA_BYTES = 8 * 1024; // 8K

    private final ManagedChannel keepAliveChannel;
    private final DaprGrpc.DaprStub keepAliveClient;
    private final ManagedChannel blockingChannel;
    private final DaprGrpc.DaprBlockingStub blockingClient;
    private final int timeoutMillis;

    public enum State {
        AVAILABLE, UNAVAILABLE
    }

    /**
     * @param host
     * @param port
     */
    public DaprClient(String host, int port) {
        this(host, port, DEFAULT_TIMEOUT);
    }

    /**
     * @param host
     * @param port
     * @param timeoutMillis
     */
    public DaprClient(String host, int port, int timeoutMillis) {
        this(host, port, timeoutMillis, DEFAULT_MAX_MESSAGE_BYTES, DEFAULT_MAX_METADATA_BYTES);
    }


    /**
     * @param host
     * @param port
     * @param timeoutMillis
     * @param maxMessageBytes
     * @param maxMetadataBytes
     */
    public DaprClient(String host, int port, int timeoutMillis, int maxMessageBytes, int maxMetadataBytes) {
        // for keepalive.
        this.keepAliveChannel = ManagedChannelBuilder.forAddress(host, port)
                .usePlaintext()
                .maxInboundMessageSize(maxMessageBytes)
                .maxInboundMetadataSize(maxMetadataBytes)
                .intercept(new DaprClientContextInterceptor())
                .userAgent("sdk-java/v1")
                .build();
        this.keepAliveClient = DaprGrpc.newStub(keepAliveChannel);
        // for client calls.
        this.blockingChannel = ManagedChannelBuilder.forAddress(host, port)
                .usePlaintext()
                .maxInboundMessageSize(maxMessageBytes)
                .maxInboundMetadataSize(maxMetadataBytes)
                .intercept(new DaprClientContextInterceptor())
                .userAgent("sdk-java/v1")
                .build();
        this.blockingClient = DaprGrpc.newBlockingStub(blockingChannel);
        this.timeoutMillis = timeoutMillis;
    }

    /**
     * @return
     */
    public State state() {
        ConnectivityState connState = this.blockingChannel.getState(false);
        return (connState == ConnectivityState.READY || connState == ConnectivityState.IDLE) ? State.AVAILABLE : State.UNAVAILABLE;
    }

    /**
     * @param action
     * @param <T>
     * @return
     */
    public <T> T execute(DaprAction<T> action) {
        Objects.requireNonNull(action, "action must not be null.");
        try {
            return action.invoke(this);
        } catch (Throwable e) {
            throw DaprException.fromThrowable(e);
        }
    }

    /**
     * @param request
     * @param responseObserver
     */
    public void abolishService(DaprProtos.AbolishRequest request, StreamObserver<DaprProtos.AbolishResponse> responseObserver) {
        keepAliveClient.abolishService(request, responseObserver);
    }

    /**
     * @param request
     * @param responseObserver
     */
    public void establishService(DaprProtos.EstablishRequest request, StreamObserver<DaprProtos.EstablishResponse> responseObserver) {
        keepAliveClient.establishService(request, responseObserver);
    }

    /**
     * @param responseObserver
     * @return
     */
    public StreamObserver<DaprProtos.KeepAliveRequest> keepAliveService(StreamObserver<DaprProtos.KeepAliveResponse> responseObserver) {
        return keepAliveClient.keepAliveService(responseObserver);
    }

    /**
     * <pre>
     * Invokes a method on a remote Dapr app.
     * </pre>
     *
     * @param request
     */
    public DaprCommonProtos.InvokeResponse invokeService(DaprCommonProtos.InvokeService request) {
        return blockingClient.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS).invokeService(request);
    }

    /**
     * <pre>
     * Invokes binding data to specific output bindings
     * </pre>
     *
     * @param request
     */
    public DaprProtos.InvokeBindingResponse invokeBinding(DaprProtos.InvokeBindingRequest request) {
        return blockingClient.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS).invokeBinding(request);
    }

    /**
     * <pre>
     * Gets the state for a specific key.
     * </pre>
     *
     * @param request
     */
    public DaprProtos.GetStateResponse getState(DaprProtos.GetStateRequest request) {
        return blockingClient.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS).getState(request);
    }

    /**
     * <pre>
     * Gets a bulk of state items for a list of keys
     * </pre>
     *
     * @param request
     */
    public DaprProtos.GetBulkStateResponse getBulkState(DaprProtos.GetBulkStateRequest request) {
        return blockingClient.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS).getBulkState(request);
    }

    /**
     * <pre>
     * Deletes the state for a specific key.
     * </pre>
     *
     * @param request
     */
    public Empty deleteState(DaprProtos.DeleteStateRequest request) {
        return blockingClient.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS).deleteState(request);
    }

    /**
     * <pre>
     * Deletes a bulk of state items for a list of keys
     * </pre>
     *
     * @param request
     */
    public Empty deleteBulkState(DaprProtos.DeleteBulkStateRequest request) {
        return blockingClient.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS).deleteBulkState(request);
    }

    /**
     * <pre>
     * Saves the state for a specific key.
     * </pre>
     *
     * @param request
     */
    public Empty setState(DaprProtos.SetStateRequest request) {
        return blockingClient.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS).setState(request);
    }

    /**
     * <pre>
     * Executes transactions for a specified store
     * </pre>
     *
     * @param request
     */
    public DaprProtos.ExecuteStateResponse executeState(DaprProtos.ExecuteStateRequest request) {
        return blockingClient.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS).executeState(request);
    }

    /**
     * <pre>
     * Gets secrets from secret stores.
     * </pre>
     *
     * @param request
     */
    public DaprProtos.GetSecretResponse getSecret(DaprProtos.GetSecretRequest request) {
        return blockingClient.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS).getSecret(request);
    }

    /**
     * <pre>
     * Gets a bulk of secrets
     * </pre>
     *
     * @param request
     */
    public DaprProtos.GetBulkSecretResponse getBulkSecret(DaprProtos.GetBulkSecretRequest request) {
        return blockingClient.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS).getBulkSecret(request);
    }

    /**
     * <pre>
     * Gets metadata of the sidecar
     * </pre>
     *
     * @param request
     */
    public DaprProtos.GetMetadataResponse getMetadata(Empty request) {
        return blockingClient.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS).getMetadata(request);
    }

    /**
     * <pre>
     * Sets value in extended metadata of the sidecar
     * </pre>
     *
     * @param request
     */
    public Empty setMetadata(DaprProtos.SetMetadataRequest request) {
        return blockingClient.withDeadlineAfter(timeoutMillis, TimeUnit.MILLISECONDS).setMetadata(request);
    }

    @Override
    public void close() {
        try {
            keepAliveChannel.shutdownNow();
        } catch (Exception ignore) {
        }

        try {
            blockingChannel.shutdownNow();
        } catch (Exception ignore) {
        }
    }
}
